package cecClient

import (
	"fmt"
	cloudevents "github.com/cloudevents/sdk-go/v2"
	"github.com/nats-io/nats.go"
	"reflect"
	"strings"
	"time"
	"unsafe"
)

const (
	cecNatsUrl          = "nats"
	cecDefaultJetStream = "cecstream."
)

type CecClient struct {
	nc *nats.Conn
}

// eventHandler is a callback function that processes messages delivered to
// asynchronous subscribers.
type eventHandler func(event *cloudevents.Event)

//connect with retry
//func Connect(url string) (*CecClient, error) {
//	con, err1 := nats.Connect(cecNatsUrl,
//		nats.RetryOnFailedConnect(true),
//		nats.MaxReconnects(10),
//		nats.ReconnectWait(time.Second),
//		nats.ReconnectHandler(func(_ *nats.Conn) {
//			//TODO: Note that this will be invoked for the first asynchronous connect.
//		}))
//
//	if err1 != nil {
//		return nil, err1
//	}
//	return &CecClient{con}, nil
//}

func Connect(url string) (conn *nats.Conn, err error) {
	con, err1 := nats.Connect(url,
		nats.RetryOnFailedConnect(true),
		nats.MaxReconnects(-1),
		nats.ReconnectWait(time.Second),
		nats.ReconnectHandler(func(_ *nats.Conn) {
			fmt.Printf("NATS got reconnected !")
		}))

	if err1 != nil {
		return nil, err1
	}
	return con, nil
}

// 生产者发送结束后关闭连接，下次需要发送事件时再连接
func Publish(event cloudevents.Event, url ...string) error {
	if err := event.Validate(); err != nil {
		return err
	}

	cc := &nats.Conn{}
	var err error
	if url != nil && len(url) > 0 {
		cc, err = Connect(url[0])
	} else {
		cc, err = Connect(cecNatsUrl)
	}
	if err != nil {
		return fmt.Errorf("和nats server建立连接失败:%v", err)
	}
	defer cc.Close()

	js, err1 := cc.JetStream()
	if err1 != nil {
		return err1
	}

	err2 := declareStream(js)
	if err2 != nil {
		return err2
	}

	msg := nats.Msg{
		Data:   event.Data(),
		Header: map[string][]string{},
	}
	msg.Header.Add("content-type", event.DataContentType())
	msg.Header.Add("specversion", event.SpecVersion())
	msg.Header.Add("time", event.Time().Format(time.RFC3339))
	msg.Header.Add("id", event.ID())
	msg.Header.Add("source", event.Source())
	msg.Header.Add("type", event.Type())

	for k, v := range event.Extensions() {
		msg.Header.Add(k, v.(string))
	}
	//cecstream.region1.ecs.create
	msg.Subject = cecDefaultJetStream + event.Subject()

	_, err3 := js.PublishMsg(&msg)
	if err3 != nil {
		return err3
	}
	return nil
}

// 消费者订阅后不关闭连接，有事件直接消费
// Subject subscriptions in NATS may be either 'individual' subscriptions or queue group subscriptions
func Subscribe(filter, queue string, eh eventHandler, url ...string) error {
	cc := &nats.Conn{}
	var err error
	if url != nil && len(url) > 0 {
		cc, err = Connect(url[0])
	} else {
		cc, err = Connect(cecNatsUrl)
	}
	if err != nil {
		return fmt.Errorf("和nats server建立连接失败:%v", err)
	}

	js, err1 := cc.JetStream()
	if err1 != nil {
		return err1
	}

	err2 := declareStream(js)
	if err2 != nil {
		return err2
	}

	que := strings.ReplaceAll(strings.ReplaceAll(strings.ReplaceAll(queue, ".", "0"), "*", "1"), ">", "2")
	queName := que + strings.ReplaceAll(strings.ReplaceAll(strings.ReplaceAll(filter, ".", "0"), "*", "1"), ">", "2")
	_, err3 := js.QueueSubscribe(cecDefaultJetStream+filter, queName, func(m *nats.Msg) {
		convertToEvent(m, eh)
		m.Ack()
	}, nats.ManualAck())
	if err3 != nil {
		return err3
	}
	return nil
}

//func (cc *CecClient) Subscribe(filter string, eh eventHandler) error {
//	js, err1 := cc.nc.JetStream()
//	if err1 != nil {
//		return err1
//	}
//
//	_, err2 := js.Subscribe(cecDefaultJetStream+filter, func(m *nats.Msg) {
//		convertToEvent(m, eh)
//		m.Ack()
//	}, nats.Durable("eventconsumer2"), nats.ManualAck())
//	if err2 != nil {
//		return err2
//	}
//	return nil
//}

func convertToEvent(msg *nats.Msg, eh eventHandler) *cloudevents.Event {
	event := cloudevents.NewEvent(cloudevents.VersionV1)
	event.SetDataContentType(msg.Header.Get("content-type"))
	event.SetSpecVersion(msg.Header.Get("specversion"))
	ts, err := time.Parse(time.RFC3339, msg.Header.Get("time"))
	if err != nil {
		fmt.Println("convert timestamp error:", err)
	}
	event.SetTime(ts)
	event.SetID(msg.Header.Get("id"))
	event.SetSource(msg.Header.Get("source"))
	event.SetType(msg.Header.Get("type"))
	event.SetSubject(strings.TrimPrefix(msg.Subject, cecDefaultJetStream))
	event.SetExtension("region", msg.Header.Get("region"))
	event.SetExtension("cluster", msg.Header.Get("cluster"))
	event.SetExtension("productcode", msg.Header.Get("productcode"))
	event.SetExtension("resouce", msg.Header.Get("resouce"))
	event.SetExtension("accountid", msg.Header.Get("accountid"))
	event.SetData(msg.Header.Get("content-type"), msg.Data)
	eh(&event)
	return &event
}

func declareStream(js nats.JetStreamContext) error {
	_, err := js.StreamInfo(strings.TrimSuffix(string(cecDefaultJetStream), "."))
	if err != nil {
		_, err = js.AddStream(&nats.StreamConfig{
			Name:      strings.TrimSuffix(string(cecDefaultJetStream), "."),
			Subjects:  []string{string(cecDefaultJetStream) + ">"},
			Retention: nats.LimitsPolicy,
		})
		if err != nil {
			return err
		}
	}
	return nil
}

var sizeOfMyStruct = int(unsafe.Sizeof(cloudevents.Event{}))

// cloud event 结构体转为 byte 数组
func StructToBytes(s *cloudevents.Event) []byte {
	var x reflect.SliceHeader
	x.Len = sizeOfMyStruct
	x.Cap = sizeOfMyStruct
	x.Data = uintptr(unsafe.Pointer(s))
	return *(*[]byte)(unsafe.Pointer(&x))
}

// byte 数组转为 cloud event 结构体
func BytesToStruct(b []byte) *cloudevents.Event {
	return (*cloudevents.Event)(unsafe.Pointer(
		(*reflect.SliceHeader)(unsafe.Pointer(&b)).Data,
	))
}
